#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <pthread.h>
#include <stdbool.h>
#include "MQTTClient.h"
#include "cJSON.h"
#include "msg_queue_peer.h"
#include "shmem.h"
#include "list.h"


#define ADDRESS     "tcp://192.168.38.187:1883"
#define SUB_CLIENTID    "ExampleClientSub"
#define PUB_CLIENTID    "ExampleClientPub"
#define SUB_TOPIC       "down"
#define PUB_TOPIC       "up"
#define QOS         1
#define TIMEOUT     10000L

struct share_mod 
{
    bool door;
    float tem;
    float weight;
    float in_pre;
    float air_pre;
    float in_h2;
    float out_h2;
};

struct share_stm
{
    bool gate;
    bool gs;
    bool LD1;
    bool LD3;
    bool FAN;
    bool FIRE;
    float temp;
    float humi;
    float Vbat;   
};

struct share
{
    int id;
    struct share_mod mod;
    struct share_stm stm;
};


void mqtt_init();
int json_init();
void shmem_init();
struct shm_param *share_init();
void msg_init();
void up_data();
void down_data(int key,int val);
void *pthread_up(void *arg);
void *pthread_sql(void *arg);
//void *pthread_dir(void *arg);
void struct_fz(struct share temp,struct share *share_addr);
bool struct_pd(struct share temp,struct share *share_addr);
bool struct_pd_equ(struct share temp,struct share *share_addr);
void delivered(void *context, MQTTClient_deliveryToken dt);
int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message);
void connlost(void *context, char *cause);



int type;//１刷新上报 2间隔上报 3历史记录 4控制设备
int period;
int rc;
int ch;
struct shm_param para;
struct share *share_addr;
struct share share_temp;
int reflash;
struct msg_node msg={0,0,"111",0,0};

volatile MQTTClient_deliveryToken deliveredtoken;
MQTTClient client_sub;
MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
MQTTClient client_pub;
MQTTClient_connectOptions conn_opts2 = MQTTClient_connectOptions_initializer;
MQTTClient_message pubmsg = MQTTClient_message_initializer;
MQTTClient_deliveryToken token;


int main(int argc, char const *argv[])
{
    mqtt_init();
    share_init();
    struct_fz(share_temp,share_addr);
    json_init();
    
    pthread_t tid;
    pthread_t tid2;
    // pthread_t tid3;
    if(pthread_create(&tid,NULL,pthread_up,NULL))
    {
        printf("pthread_up creat err");
        return -1;
    }
    if(pthread_create(&tid2,NULL,pthread_sql,NULL))
    {
        printf("pthread_up creat err");
        return -1;
    }
    // if(pthread_create(&tid3,NULL,pthread_dir,NULL))
    // {
    //     printf("pthread_up creat err");
    //     return -1;
    // }
    do 
    {
        ch = getchar();
    } while(ch!='Q' && ch != 'q');

    shm_del(&para);
    pthread_detach(tid);
    pthread_detach(tid2);
    // pthread_detach(tid3);
    MQTTClient_unsubscribe(client_sub, SUB_TOPIC);
    MQTTClient_disconnect(client_sub, 10000);
    MQTTClient_destroy(&client_sub);
    //结束发布客户端
    MQTTClient_disconnect(client_pub, 10000);
    MQTTClient_destroy(&client_pub);
  
    return 0;
}


void mqtt_init()
{
    

    MQTTClient_create(&client_sub, ADDRESS, SUB_CLIENTID,
        MQTTCLIENT_PERSISTENCE_NONE, NULL);
    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;

    //订阅接收
    MQTTClient_setCallbacks(client_sub, NULL, connlost, msgarrvd, delivered);

    if ((rc = MQTTClient_connect(client_sub, &conn_opts)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to connect, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }
    printf("Subscribing to topic %s\nfor client %s using QoS%d\n\n"
           "Press Q<Enter> to quit\n\n", SUB_TOPIC, SUB_CLIENTID, QOS);
    MQTTClient_subscribe(client_sub, SUB_TOPIC, QOS);

    //发布客户端


    MQTTClient_create(&client_pub, ADDRESS, PUB_CLIENTID,
        MQTTCLIENT_PERSISTENCE_NONE, NULL);
    conn_opts2.keepAliveInterval = 20;
    conn_opts2.cleansession = 1;

    if ((rc = MQTTClient_connect(client_pub, &conn_opts2)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to connect, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }
}

int json_init()
{
    FILE *fp = fopen("./config.json", "r");
    if(fp == NULL)
    {
        perror("fopen err");
        return -1;
    }
    char buf[2048] = "";
    fread(buf, 2048, 1, fp);
    cJSON *root=cJSON_Parse(buf);
    //printf("----------\n");
    cJSON *item=cJSON_GetObjectItem(root,"version");
    //printf("%s:%s\n",item->string,item->valuestring);

    cJSON *report=cJSON_GetObjectItem(root,"report");
    item=cJSON_GetObjectItem(report,"type");
    type=item->valueint;
    //printf("%s:%d\n",item->string,item->valueint);
    item=cJSON_GetObjectItem(report,"period");
    period=item->valueint;
    //printf("%s:%d\n",item->string,item->valueint);

    cJSON *stm32=cJSON_GetObjectItem(root,"stm32");
    cJSON *data=cJSON_GetObjectItem(stm32,"data");
    for(int i=0;i<cJSON_GetArraySize(data);i++)
    {
        item=cJSON_GetArrayItem(data,i);
        cJSON *key=cJSON_GetObjectItem(item,"key");
        cJSON *name=cJSON_GetObjectItem(item,"name");
        cJSON *type=cJSON_GetObjectItem(item,"type");
        //printf("%s%d:%d\n",key->string,i,key->valueint);
        //printf("%s%d:%d\n",name->string,i,name->valueint);
        //printf("%s%d:%d\n",type->string,i,type->valueint);
    }
    //printf("\n");
   
    cJSON *modbus=cJSON_GetObjectItem(root,"modbus");
    data=cJSON_GetObjectItem(modbus,"data");
    for(int i=0;i<cJSON_GetArraySize(data);i++)
    {
        item=cJSON_GetArrayItem(data,i);
        cJSON *key=cJSON_GetObjectItem(item,"key");
        cJSON *name=cJSON_GetObjectItem(item,"name");
        cJSON *type=cJSON_GetObjectItem(item,"type");
        cJSON *addr=cJSON_GetObjectItem(item,"addr");
        //printf("%s%d:%d\n",key->string,i,key->valueint);
        //printf("%s%d:%d\n",name->string,i,name->valueint);
        //printf("%s%d:%d\n",addr->string,i,addr->valueint);
        //printf("%s%d:%d\n",type->string,i,type->valueint);
    }
    

    fclose(fp);

    return 0;
}

struct shm_param *share_init()
{
    if(shm_init(&para,"shm",1024)<0)
    {
        printf("shm init err\n");
        return NULL;
    }
    share_addr=(struct share *)shm_getaddr(&para);
    if(share_addr==NULL)
    {
        printf("shm get addr err\n");
        return NULL;
    }

    printf("%p %.2f\n",share_addr,share_addr->mod.tem);
}

void up_data()
{
        cJSON *root =cJSON_CreateObject();

        cJSON *itemsg=cJSON_CreateObject();
        cJSON_AddNumberToObject(itemsg,"door",share_addr->mod.door);
        cJSON_AddNumberToObject(itemsg,"tem",share_addr->mod.tem);
        cJSON_AddNumberToObject(itemsg,"weight",share_addr->mod.weight);
        cJSON_AddNumberToObject(itemsg,"in_pre",share_addr->mod.in_pre);
        cJSON_AddNumberToObject(itemsg,"air_pre",share_addr->mod.air_pre);
        cJSON_AddNumberToObject(itemsg,"in_h2",share_addr->mod.in_h2);
        cJSON_AddNumberToObject(itemsg,"out_h2",share_addr->mod.out_h2);
        cJSON_AddItemToObject(root,"modbus", itemsg);

        cJSON *item2=cJSON_CreateObject();
        cJSON_AddNumberToObject(item2,"gate",share_addr->stm.gate);
        cJSON_AddNumberToObject(item2,"gs",share_addr->stm.gs);
        cJSON_AddNumberToObject(item2,"LD1",share_addr->stm.LD1);
        cJSON_AddNumberToObject(item2,"LD3",share_addr->stm.LD3);
        cJSON_AddNumberToObject(item2,"FAN",share_addr->stm.FAN);
        cJSON_AddNumberToObject(item2,"FIRE",share_addr->stm.FIRE);
        cJSON_AddNumberToObject(item2,"temp",share_addr->stm.temp);
        cJSON_AddNumberToObject(item2,"humi",share_addr->stm.humi);
        cJSON_AddNumberToObject(item2,"Vbat",share_addr->stm.Vbat);
        cJSON_AddItemToObject(root,"stm32", item2);

        char *val=cJSON_Print(root);
        char *pub_payload;
        char pub_buffer[1024]={};
        pub_payload=cJSON_Print(root);

        pubmsg.payload = pub_payload;
        pubmsg.payloadlen = (int)strlen(pub_payload);
        pubmsg.qos = QOS;
        pubmsg.retained = 0;
        MQTTClient_publishMessage(client_pub, PUB_TOPIC, &pubmsg, &token);
        //printf("Waiting for up to %d seconds for publication of %s\n"
           // "on topic %s for client with ClientID: %s\n",
        //(int)(TIMEOUT/1000), pub_payload, PUB_TOPIC, PUB_CLIENTID);
        rc = MQTTClient_waitForCompletion(client_pub, token, TIMEOUT);
        //printf("Message with delivery token %d delivered\n", token);
        printf("已上传%d条数据\n", token);
}

void down_data(int key,int val)
{
    if(key>=200&&key<300)
    {
        struct msg_node msg={888,key,"111",0,val};
        printf("消息队列:  key:%d  name:%s  addr:%d  data:%d\n", msg.key, msg.name, msg.addr, msg.data);
        msg_queue_send("msg", &msg, sizeof(msg), 0);
        printf("已向设备%d发送更改状态为%d\n",key,val);
        
    }
    else if(key>=300&&key<400)
    {
        struct msg_node msg={999,key,"111",0,val};
        printf("消息队列:  key:%d  name:%s  addr:%d  data:%d\n", msg.key, msg.name, msg.addr, msg.data);
        msg_queue_send("msg", &msg, sizeof(msg), 0);
        printf("已向设备%d发送更改状态为%d\n",key,val);
    }
    
}

//上报数据
void *pthread_up(void *arg)
{
    while(1)
    {
        if(type==1)
        {
            // if(struct_pd_equ(share_temp,share_addr))
            //  {
            //     up_data();
            //     //sleep(1);
            //  }
            //printf("reflash:%d\n",reflash);
            if(reflash==1)
            up_data();
            reflash=0;
            sleep(0.1);
        }
        else if(type==2)
        {
            // if(struct_pd_equ(share_temp,share_addr))
            // {
            // up_data();
            //     //sleep(1);
            // }
            up_data();
            sleep(period);
        }
        else if(type==3)
        {
             if(struct_pd(share_temp,share_addr))
             {
                up_data();
                //sleep(1);
             }
        
        }
        if(struct_pd_equ(share_temp,share_addr))
        {
            up_data();
            //sleep(1);
        }
        struct_fz(share_temp,share_addr);
        //printf("%d\n",type);
    }
}

//数据库
void *pthread_sql(void *arg)
{
   
}

//处理指令
// void *pthread_dir(void *arg)
// {

// }

void struct_fz(struct share temp,struct share *share_addr)
{
    // temp.mod=share_addr->mod;
    // temp.stm=share_addr->stm;

    share_temp.mod.air_pre=share_addr->mod.air_pre;
    share_temp.mod.door=share_addr->mod.door;
    share_temp.mod.in_h2=share_addr->mod.in_h2;
    share_temp.mod.in_pre=share_addr->mod.in_pre;
    share_temp.mod.out_h2=share_addr->mod.out_h2;
    share_temp.mod.tem=share_addr->mod.tem;
    share_temp.mod.weight=share_addr->mod.weight;
    share_temp.stm.FAN=share_addr->stm.FAN;
    share_temp.stm.FIRE=share_addr->stm.FIRE;
    share_temp.stm.gate=share_addr->stm.gate;
    share_temp.stm.gs=share_addr->stm.gs;
    share_temp.stm.humi=share_addr->stm.humi;
    share_temp.stm.LD1=share_addr->stm.LD1;
    share_temp.stm.LD3=share_addr->stm.LD3;
    share_temp.stm.temp=share_addr->stm.temp;
    share_temp.stm.Vbat=share_addr->stm.Vbat;

    //printf("%f\n",temp.mod.air_pre);
}

bool struct_pd(struct share temp,struct share *share_addr)
{
    if(share_temp.mod.air_pre!=share_addr->mod.air_pre
    ||share_temp.mod.door!=share_addr->mod.door
    ||share_temp.mod.in_h2!=share_addr->mod.in_h2
    ||share_temp.mod.in_pre!=share_addr->mod.in_pre
    ||share_temp.mod.out_h2!=share_addr->mod.out_h2
    ||share_temp.mod.tem!=share_addr->mod.tem
    ||share_temp.mod.weight!=share_addr->mod.weight
    )
    {
        return TRUE;
    }
    else if(share_temp.stm.FAN!=share_addr->stm.FAN
            ||share_temp.stm.FIRE!=share_addr->stm.FIRE
            ||share_temp.stm.gate!=share_addr->stm.gate
            ||share_temp.stm.gs!=share_addr->stm.gs
            ||share_temp.stm.humi!=share_addr->stm.humi
            ||share_temp.stm.LD1!=share_addr->stm.LD1
            ||share_temp.stm.LD3!=share_addr->stm.LD3
            ||share_temp.stm.temp!=share_addr->stm.temp
            ||share_temp.stm.Vbat!=share_addr->stm.Vbat
            )
            {
                return TRUE;
            }
    else
    {
        return FALSE;
    }
    //return memcmp(&share_temp,share_addr,sizeof(share_temp));
}


bool struct_pd_equ(struct share temp,struct share *share_addr)
{
    if(share_temp.mod.door!=share_addr->mod.door
    )
    {
        return TRUE;
    }
    else if(share_temp.stm.FAN!=share_addr->stm.FAN
            ||share_temp.stm.FIRE!=share_addr->stm.FIRE
            ||share_temp.stm.gate!=share_addr->stm.gate
            ||share_temp.stm.gs!=share_addr->stm.gs
            ||share_temp.stm.LD1!=share_addr->stm.LD1
            ||share_temp.stm.LD3!=share_addr->stm.LD3
            )
            {
                return TRUE;
            }
    else
    {
        return FALSE;
    }
    //return memcmp(&share_temp,share_addr,sizeof(share_temp));
}

void delivered(void *context, MQTTClient_deliveryToken dt)
{
    printf("Message with token value %d delivery confirmed\n", dt);
    deliveredtoken = dt;
}

int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message)
{
    int i;
    char* payloadptr;


    //printf("Message arrived\n");
    //printf("     topic: %s\n", topicName);
    //printf("   message: ");

    payloadptr = message->payload;
    cJSON *root=cJSON_Parse(payloadptr);
    cJSON *item=cJSON_GetObjectItem(root,"type");
    if(item->valueint>=1&&item->valueint<=3)
    {
        // FILE *fp = fopen("./config.json", "r");
        // if(fp == NULL)
        // {
        //     perror("fopen err");
        //     return -1;
        // }
        // char buf[2048] = "";
        // fread(buf, 2048, 1, fp);
        // fclose(fp);
         if(type!=item->valueint)
         {
            printf("\n");
             printf("现在是模式%d\n",type);
             type=item->valueint;
             printf("已切换至模式%d\n",type);
        //     FILE *fp = fopen("./config.json", "w+");
        //     if(fp == NULL)
        //     {
        //         perror("fopen err");
        //         return -1;
        //     }
        //     cJSON *root1=cJSON_Parse(buf);
        //     cJSON *report=cJSON_GetObjectItem(root1,"report");
        //     cJSON *item3=cJSON_GetObjectItem(report,"type");
        //     cJSON_SetNumberValue(item3,item->valueint);
        //     //printf("%s\n",buf);
        //     fwrite(buf,sizeof(char),strlen(buf),fp);
        //     fclose(fp);
         }
    
        if(item->valueint==1)
        {
            //cJSON *itemsg=cJSON_GetObjectItem(root,"limit");
            //if(itemsg->valuestring=="all")
            //{
                reflash=1;
            //}
        }
    }
    else if(item->valueint==4)
    {
        cJSON *key=cJSON_GetObjectItem(root,"key");
        cJSON *val=cJSON_GetObjectItem(root,"val");
        down_data(key->valueint,val->valueint);
    }

    //for(i=0; i<message->payloadlen; i++)
    //{
        //putchar(*payloadptr++);
   // }
    putchar('\n');
    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);
    return 1;
}

void connlost(void *context, char *cause)
{
    printf("\nConnection lost\n");
    printf("     cause: %s\n", cause);
}
